
/*
 * HadoopMapper.java
 *
 * Created on May 6, 2012, 10:27:56 PM
 */

package edges;


import java.io.IOException;
// import org.apache.commons.logging.Log;
// import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;

import BasicTypes.Constants;
import BasicTypes.Position;
import BasicTypes.Table;

/**
 *
 * @author crazymooner
 */
public class HadoopMapper extends MapReduceBase implements Mapper<LongWritable,Text,Text,Text> {
    // The Karmasphere Studio Workflow Log displays logging from Apache Commons Logging, for example:
    // private static final Log LOG = LogFactory.getLog("edges.HadoopMapper");

	private int GetNext(double test) {
		double fromNetID = 0.925;
		double desiredDensity = 0.59;
		double wMin = 0.4 * fromNetID;
		double wLimit = wMin + desiredDensity;
		return (((test >= wMin) && (test < wLimit)) ? 1 : 0);
	}
	
	
    @Override
    public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter)
            throws IOException {
    	if(GetNext(Double.parseDouble(value.toString())) == 1){
    		int index = (int) (key.get()/Constants.KEYSIZE);
        	Position p = Table.getPosition(index);
        	if(p.y > 0){
        		Position below = new Position(p.x,p.y - 1);
        		output.collect(new Text(below.getString()), new Text(p.getString()));
        		if(p.x > 0){
            		Position belowleft = new Position(p.x - 1,p.y - 1);
            		output.collect(new Text(belowleft.getString()), new Text(p.getString()));
        		}
        		if(p.x < Constants.SIZE - 1){
            		Position belowright = new Position(p.x + 1,p.y - 1);
            		output.collect(new Text(belowright.getString()), new Text(p.getString()));
        		}
        	}
        	if(p.x > 0){
        		Position left = new Position(p.x -1 , p.y);
        		output.collect(new Text(left.getString()), new Text(p.getString()));
        	}
        	
        	output.collect(new Text(p.getString()), new Text(p.getString()));
    	}
    }
}
